package com.bestcem.xm.deliver.config;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskDecorator;
import org.springframework.lang.NonNull;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.*;

/**
 * @ClassName: DeliverExecutorConfig
 * @Description: 其余投放的异步任务，抄一下
 * @Author: chenglong.yue
 * @Date: 2022/11/25 18:18
 */
@Configuration
@EnableAsync
@Slf4j
@RequiredArgsConstructor
public class DeliverExecutorConfig implements AsyncConfigurer {

    private @NonNull XmCustomThreadFactory factory;
    /**
     * 核心线程数
     * 默认的核心线程数为1
     */
    private static final int CORE_POOL_SIZE = 10;
    /**
     * 最大线程数
     * 默认的最大线程数是Integer.MAX_VALUE
     */
    private static final int MAX_POOL_SIZE = 50;
    /**
     * 缓冲队列数
     * 默认的缓冲队列数是Integer.MAX_VALUE
     */
    private static final int QUEUE_CAPACITY = 100;

    /**
     * 允许线程空闲时间
     * 默认的线程空闲时间为60秒
     */
    private static final int KEEP_ALIVE_SECONDS = 30;

    @Bean
    @Override
    public Executor getAsyncExecutor() {//默认线程池
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(CORE_POOL_SIZE);
        taskExecutor.setMaxPoolSize(MAX_POOL_SIZE);
        taskExecutor.setQueueCapacity(QUEUE_CAPACITY);
        taskExecutor.setKeepAliveSeconds(KEEP_ALIVE_SECONDS);
        taskExecutor.setThreadNamePrefix("Task_Deliver_Spring_Default_Async_");
        taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
        taskExecutor.setAwaitTerminationSeconds(5);
        taskExecutor.setRejectedExecutionHandler(new MyRejectedExecutionHandler());
        taskExecutor.setTaskDecorator(new AfterExecuteTaskDecorator());
        //taskExecutor.initialize();
        return taskExecutor;
    }

    @Bean("wechatmpTaskExecutor")
    public ThreadPoolTaskExecutor wechatmpTaskExecutor() {//公众号投放线程池
        //ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        ThreadPoolTaskExecutor taskExecutor = new VisiableThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(CORE_POOL_SIZE);
        taskExecutor.setMaxPoolSize(MAX_POOL_SIZE);
        taskExecutor.setQueueCapacity(QUEUE_CAPACITY);
        taskExecutor.setKeepAliveSeconds(KEEP_ALIVE_SECONDS);
        taskExecutor.setThreadNamePrefix("Task_Deliver_Wechatmp_Async_");
        // 当任务完成后，长时间无待处理任务时，销毁线程池
        taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
        taskExecutor.setAwaitTerminationSeconds(5);
        taskExecutor.setThreadFactory(factory);
        //当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize，如果还有任务到来就会采取任务拒绝策略
        //通常有以下四种策略：
        //ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
        //ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
        //ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
        //ThreadPoolExecutor.CallerRunsPolicy：重试添加当前的任务，自动重复调用 execute() 方法，直到成功
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        taskExecutor.setTaskDecorator(new AfterExecuteTaskDecorator());
        // 初始化
        //通过看源码我们看到ThreadPoolTaskExecutor类（实际上是其父类）实现了InitializingBean接口，
        //则在spring容器初始化时通过父类实现的afterPropertiesSet()方法最终触发ThreadPoolTaskExecutor中的initializeExecutor方法
        //从而this.threadPoolExecutor = executor;将成员变量中的ThreadPoolExecutor实例化.
        //taskExecutor.initialize();
        return taskExecutor;
    }

    // 无返回值方法异常时，走下面逻辑
    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return (ex, method, params) -> log.error("线程池执行任务发送未知错误,执行方法：{}", method.getName(), ex);
    }

    // 有返回值方法异常时，走下面逻辑
    private static class AfterExecuteTaskDecorator implements TaskDecorator {
        @NonNull
        @Override
        public Runnable decorate(@NonNull Runnable runnable) {
            return () -> {
                try {
                    runnable.run();
                    if (runnable instanceof Future<?> && ((Future<?>) runnable).isDone()) {
                        ((Future<?>) runnable).get();
                    }
                } catch (InterruptedException e) {
                    log.error("ThreadPoolTaskExecutor异步方法{}执行被interrupt", runnable.getClass().getSimpleName(), e);
                    Thread.currentThread().interrupt();
                } catch (Exception e) {
                    log.error("ThreadPoolTaskExecutor异步方法{}执行出错", runnable.getClass().getSimpleName(), e);
                }
            };
        }
    }

    private static class MyRejectedExecutionHandler implements RejectedExecutionHandler {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            String msg = String.format("thread pool is exhausted, Reject Task Name:%s, " +
                            "Pool:%d(active:%d, core:%d, max:%d, largest:%d), " +
                            "Task:%d(completed:%d), " +
                            "Queue:%d(size:%d, remaining:%d), " +
                            "Status:(isShutdown:%s, isTerminated:%s, isTerminating:%s)",
                    r.toString(),
                    executor.getPoolSize(), executor.getActiveCount(), executor.getCorePoolSize(), executor.getMaximumPoolSize(), executor.getLargestPoolSize(),
                    executor.getTaskCount(), executor.getCompletedTaskCount(),
                    executor.getQueue().size() + executor.getQueue().remainingCapacity(), executor.getQueue().size(), executor.getQueue().remainingCapacity(),
                    executor.isShutdown(), executor.isTerminated(), executor.isTerminating());
            log.warn(msg);
            try {
                // 最大努力执行任务型，当触发拒绝策略时，在尝试一分钟的时间重新将任务塞进任务队列，当一分钟超时还没成功时，做业务操作（记录异常、报警处理等）
                if (executor.getQueue().offer(r, 1, TimeUnit.MINUTES)) {
                    log.error("等待一分钟还是不行，拜拜咯");
                }
            } catch (InterruptedException ignored) {
                Thread.currentThread().interrupt();
            }
        }
    }
}
